package represent.expression;

import java.util.*;

import metadata.schema.Schema;

import represent.Plan;
import represent.Scan;
import represent.type.Type;

/**
 * A predicate is a Boolean combination of terms.
 * 
 * @author Edward Sciore
 * 
 */

public class Predicate {
	public static final int AND = 0, OR = 1;
	private static final String[] relations = { "AND", "OR" };
	// private List<Term> terms = new ArrayList<Term>();
	private TermPair terms;

	/**
	 * Creates an empty predicate, corresponding to "true".
	 */
	public Predicate() {
		terms = null;
	}

	/**
	 * Creates a predicate containing a single term.
	 * 
	 * @param t
	 *            the term
	 */
	public Predicate(Term t) {
		terms = new TermPair(t);
	}

	/**
	 * Modifies the predicate to be the conjunction of itself and the specified
	 * predicate.
	 * 
	 * @param pred
	 *            the other predicate
	 */
	public void conjoinWith(Predicate pred, int rel) {
		terms = new TermPair(terms, pred.terms, rel);
	}

	/**
	 * Returns true if the predicate evaluates to true with respect to the
	 * specified scan.
	 * 
	 * @param s
	 *            the scan
	 * @return true if the predicate is true in the scan
	 */
	public boolean isSatisfied(Scan s) {
		return terms == null || terms != null && terms.isSatisfied(s);
	}

	/**
	 * Calculates the extent to which selecting on the predicate reduces the
	 * number of records output by a query. For example if the reduction factor
	 * is 2, then the predicate cuts the size of the output in half.
	 * 
	 * @param p
	 *            the query's plan
	 * @return the integer reduction factor.
	 */
	public int reductionFactor(Plan p) {
		if (terms == null)
			return 1;
		return terms.reductionFactor(p);
	}

	/**
	 * Returns the subpredicate that applies to the specified schema.
	 * 
	 * @param sch
	 *            the schema
	 * @return the subpredicate applying to the schema
	 */
	public Predicate selectPred(Schema sch) {
		Predicate result = new Predicate();
		List<TermPair> cnf = predToCNF(terms);
		for (TermPair t : cnf)
			if (t.appliesTo(sch)) {
				if (result.terms == null)
					result.terms = t;
				else
					result.terms = new TermPair(result.terms, t, AND);
			}
		return result;
	}

	/**
	 * Returns the subpredicate that NOT applies to the specified schema.
	 * 
	 * @param sch
	 *            the schema
	 * @return the subpredicate NOT applying to the schema
	 */
	public Predicate unableSelectPred(Schema sch) {
		Predicate result = new Predicate();
		List<TermPair> cnf = predToCNF(terms);
		for (TermPair t : cnf)
			if (!t.appliesTo(sch)) {
				if (result.terms == null)
					result.terms = t;
				else
					result.terms = new TermPair(result.terms, t, AND);
			}
		return result;
	}

	private List<TermPair> predToCNF(TermPair terms) {
		if (terms == null)
			return new LinkedList<TermPair>();
		List<TermPair> rst = new LinkedList<TermPair>();
		if (terms.lhs != null && terms.rhs != null) {
			if (terms.rel == AND) {
				rst.addAll(predToCNF(terms.lhs));
				rst.addAll(predToCNF(terms.rhs));
			} else {
				List<TermPair> l = predToCNF(terms.lhs);
				List<TermPair> r = predToCNF(terms.rhs);
				for (TermPair tl : l)
					for (TermPair tr : r)
						rst.add(new TermPair(tl, tr, OR));
			}
		} else {
			rst.add(terms);
		}
		return rst;
	}

	public List<Predicate> predToCNF() {
		List<TermPair> t_cnf = predToCNF(terms);
		List<Predicate> p_cnf = new LinkedList<Predicate>();
		for (TermPair t : t_cnf) {
			Predicate p = new Predicate();
			p.terms = t;
			p_cnf.add(p);
		}
		return p_cnf;
	}

	/**
	 * Returns the subpredicate consisting of terms that apply to the union of
	 * the two specified schemas, but not to either schema separately.
	 * 
	 * @param sch1
	 *            the first schema
	 * @param sch2
	 *            the second schema
	 * @return the subpredicate whose terms apply to the union of the two
	 *         schemas but not either schema separately.
	 */
	public Predicate joinPred(Schema sch1, Schema sch2) {
		Predicate result = new Predicate();
		Schema newsch = new Schema();
		newsch.addAll(sch1);
		newsch.addAll(sch2);
		List<TermPair> cnf = predToCNF(terms);
		for (TermPair t : cnf) {

			if (!t.appliesTo(sch1) && !t.appliesTo(sch2) && t.appliesTo(newsch)) {
				if (result.terms == null)
					result.terms = t;
				else
					result.terms = new TermPair(result.terms, t, AND);
			}
		}
		return result;
	}

	/**
	 * Determines if there is a term of the form "F op c" where F is the
	 * specified field and c is some constant. If so, the method returns that
	 * constant. If not, the method returns null.
	 * 
	 * @param fldname
	 *            the name of the field
	 * @return either the constant or null
	 */
	public Type equatesWithConstant(String fldname) {
		if (terms == null)
			return null;
		return terms.equatesWithConstant(fldname);
	}

	/**
	 * Determines if there is a term of the form "F1 op F2" where F1 is the
	 * specified field and F2 is another field. If so, the method returns the
	 * name of that field. If not, the method returns null.
	 * 
	 * @param fldname
	 *            the name of the field
	 * @return the name of the other field, or null
	 */
	public String equatesWithField(String fldname) {
		if (terms == null)
			return null;
		return terms.equatesWithField(fldname);
	}

	/**
	 * needed fields of this predicate
	 * 
	 * @return a collection of field name
	 */
	public Collection<String> getNeededFields() {
		if (terms == null)
			return new HashSet<String>();
		else
			return terms.getNeededFields();
	}

	/**
	 * Add table name in this predicate
	 * 
	 * @param schema
	 */
	public void addTableName(Schema schema) {
		if (terms != null)
			terms.addTableName(schema);
	}
	
	/**
	 * Expend * fields in predicate
	 * @param allSchema
	 */
	public void expandAllFields(Schema upperLevelSchema) {
		if (terms != null)
			terms.expandAllFields(upperLevelSchema);
	}

	public String toString() {
		if (terms == null)
			return "";
		return terms.toString();
	}

	public boolean isEmpty() {
		return terms == null;
	}

	class TermPair {
		TermPair lhs;
		TermPair rhs;
		int rel;
		Term val;

		public TermPair(TermPair l, TermPair r, int re) {
			lhs = l;
			rhs = r;
			rel = re;
		}

		public TermPair(Term t) {
			val = t;
		}

		public boolean isSatisfied(Scan s) {
			if (lhs != null && rhs != null) {
				if (rel == AND)
					return lhs.isSatisfied(s) && rhs.isSatisfied(s);
				else if (rel == OR)
					return lhs.isSatisfied(s) || rhs.isSatisfied(s);
				else
					throw new Error(
							"fatworm.represent.Predicate.TermPair.isSatisfied");
			} else {
				return val.isSatisfied(s);
			}
		}

		public int reductionFactor(Plan p) {
			if (lhs != null && rhs != null && rel == AND)
				return lhs.reductionFactor(p) * rhs.reductionFactor(p);
			else if (lhs != null && rhs != null && rel == OR) {
				int lr = lhs.reductionFactor(p);
				int rr = rhs.reductionFactor(p);
				return lr * rr / (lr + rr - 1);
			} else
				return val.reductionFactor(p);
		}

		public Type equatesWithConstant(String fldname) {
			if (lhs != null && rhs != null) {
				Type c = lhs.equatesWithConstant(fldname);
				if (c != null)
					return c;
				else
					return rhs.equatesWithConstant(fldname);
			} else
				return val.equatesWithConstant(fldname);

		}

		public String equatesWithField(String fldname) {
			if (lhs != null && rhs != null) {
				String s = lhs.equatesWithField(fldname);
				if (s != null)
					return s;
				else
					return rhs.equatesWithField(fldname);
			} else
				return val.equatesWithField(fldname);
		}

		public boolean appliesTo(Schema schema) {
			if (lhs != null && rhs != null)
				return lhs.appliesTo(schema) && rhs.appliesTo(schema);
			else
				return val.appliesTo(schema);
		}

		public Collection<String> getNeededFields() {
			Collection<String> flds = new HashSet<String>();
			if (lhs != null && rhs != null) {
				flds.addAll(lhs.getNeededFields());
				flds.addAll(rhs.getNeededFields());
			} else
				flds.addAll(val.getNeededFields());
			return flds;
		}

		public void addTableName(Schema schema) {
			if (lhs != null && rhs != null) {
				lhs.addTableName(schema);
				rhs.addTableName(schema);
			} else
				val.addTableName(schema);
		}
		
		public void expandAllFields(Schema upperLevelSchema) {
			if (lhs != null && rhs != null) {
				lhs.expandAllFields(upperLevelSchema);
				rhs.expandAllFields(upperLevelSchema);
			} else
				val.expandAllFields(upperLevelSchema);
		}

		public String toString() {
			if (lhs != null && rhs != null)
				return "(" + lhs.toString() + ") " + relations[rel] + " ("
						+ rhs.toString() + ")";
			else
				return val.toString();
		}
	}
}
